package com.zhw.task;

import java.util.*;
import java.util.concurrent.*;

/**
 * 依赖任务加载的调度
 */
public class DAGScheduler {

    /**
     * 储存所有的任务
     */
    private Set<IDAGTask> mTaskSet;
    /**
     * 储存所有的任务和前置依赖任务
     */
    private Map<IDAGTask, Set<IDAGTask>> mTaskMap;
    /**
     * 存储入度为0的待执行任务
     */
    private ArrayBlockingQueue<Runnable> mTaskBlockingDeque;
    /**
     * 任务线程池
     */
    private ThreadPoolExecutor mTaskThreadPool;

    public void start(DAGProject DAGProject) {
        mTaskSet = DAGProject.getDAGTaskSet();
        mTaskMap = DAGProject.getDAGTaskMap();
        checkCircle();
        init();
        loop();
    }

    private void init() {
        int codeTaskSize = Runtime.getRuntime().availableProcessors();
        int maxTaskSize = Math.max(Runtime.getRuntime().availableProcessors() * 2 + 1, mTaskSet.size());
        int keepAliveTime = 1000;
        TimeUnit timeUnit = TimeUnit.MILLISECONDS;
        ArrayBlockingQueue<Runnable> taskBlockingDeque = new ArrayBlockingQueue<>(maxTaskSize - codeTaskSize);
        mTaskThreadPool = new ThreadPoolExecutor(codeTaskSize, maxTaskSize, keepAliveTime, timeUnit, taskBlockingDeque);
        mTaskBlockingDeque = new ArrayBlockingQueue<>(mTaskSet.size());
    }

    private void loop() {
        for (IDAGTask idagTask : mTaskSet) {
            if (idagTask.getRely() == 0) {
                mTaskBlockingDeque.add(idagTask);
            }
        }
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        singleThreadExecutor.execute(() -> {
            for (;;) {
                try {
                    while (!mTaskBlockingDeque.isEmpty()) {
                        IDAGTask executedDAGTsk = (IDAGTask) mTaskBlockingDeque.take();
                        if (executedDAGTsk.getIsAsync()) {
                            executedDAGTsk.run();
                        } else {
                            mTaskThreadPool.execute(executedDAGTsk);
                        }
                        mTaskSet.remove(executedDAGTsk);
                    }
                    if (mTaskSet.isEmpty()) {
                        singleThreadExecutor.shutdown();
                        mTaskThreadPool.shutdown();
                        return;
                    }
                    Iterator<IDAGTask> iterator = mTaskSet.iterator();
                    while (iterator.hasNext()) {
                        IDAGTask idagTask = iterator.next();
                        if (idagTask.getRely() == 0) {
                            mTaskBlockingDeque.put(idagTask);
                            iterator.remove();
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    private void checkCircle() {
        Set<IDAGTask> tempTaskSet = new HashSet<>(mTaskSet);
        Map<IDAGTask, Set<IDAGTask>> tempTaskMap = new HashMap<>();
        for (Map.Entry<IDAGTask, Set<IDAGTask>> entry:mTaskMap.entrySet()) {
            if (entry.getValue() != null) {
                Set<IDAGTask> set = new HashSet<>(entry.getValue());
                tempTaskMap.put(entry.getKey(), set);
            }
        }
        List<IDAGTask> resultTaskQueue = new LinkedList<>();
        // 储存入度为0的任务
        Queue<IDAGTask> tempTaskQueue = new LinkedList<>();
        for (IDAGTask dagTask : tempTaskSet) {
            if (tempTaskMap.get(dagTask) == null) {
                tempTaskQueue.add(dagTask);
            }
        }
        while (!tempTaskQueue.isEmpty()) {
            IDAGTask tempDAGTask = tempTaskQueue.poll();
            resultTaskQueue.add(tempDAGTask);
            for (IDAGTask dagTask : tempTaskMap.keySet()) {
                Set<IDAGTask> tempDAGSet = tempTaskMap.get(dagTask);
                if (tempDAGSet != null && tempDAGSet.contains(tempDAGTask)) {
                    tempDAGSet.remove(tempDAGTask);
                    if (tempDAGSet.size() == 0) {
                        tempTaskQueue.add(dagTask);
                    }
                }
            }
        }
        if (resultTaskQueue.size() != tempTaskSet.size()) {
            throw new IllegalArgumentException("相互依赖，玩屁啊，我不跑了！");
        }
    }

}
